<?php
/**
 * 队列Class
 *
 * $hosts参数的定义规则：
 * $hosts = array(
 * 		's_1' => array('host'=>'192.168.72.27', 'port'=>7121),
 * 		's_2' => array('host'=>'192.168.72.28', 'port'=>7121),
 * 	)
 *
 * $queue_list参数的定义规则
 * $queues = array(
 * 		'feed'   => array('server_id'=>array('s_1'));
 * 		'online' => array('server_id'=>array('s_1'));
 * 		'exp'    => array('server_id'=>array('s_2'));
 * 	)
 *
 * 配置文件里这样添加：
 * 'QUEUE' => array(
 * 		'hosts'  => $hosts,
 * 		'queues' => $queues,
 * 	)
 *
 * 所有Queue的端口号都是7000~8000之间；
 * 每个服务器需要定义一个别名，例如s_1, s_2；定义别名的好处是，当服务器地址发生变化时，别名是不变的，不会影响原有的队列数据分布
 *
 */

class Queue {

    private $app;
	private $hosts;   // 队列host
	private $q_name;  // 队列的名称
	private $real_q_name; // 真实的队列名称
	private $redis; /// redis实例
	private $connected=false; /// 标记redis是否连接

	/**
     * 构造函数
     *
     * @param string app, 每个应用的惟一编号
     * @param string q_name, 队列的名称
     * @param mixed conf, 队列的配置信息，可选
     * @access public
     */
    public function __construct($app='', $q_name='', $conf='') {

    	if (empty($app)) {
    		trigger_error(__FUNCTION__.': 没有指定app！', E_USER_ERROR);
    	}
    	if (empty($q_name)) {
    		trigger_error(__FUNCTION__.': 没有指定队列名称！', E_USER_ERROR);
    	}

    	$this->app = $app;
    	$this->q_name = $q_name;
    	$this->real_q_name = $this->app . '-' . $this->q_name;

    	if (empty($conf)) $conf = C('queue_config');

    	if (!isset($conf['queues'][$q_name])) {
    		trigger_error('队列名称['.$q_name.']在配置文件里没有定义!', E_USER_ERROR);
    	}

    	$server_ids = $conf['queues'][$q_name]['server_id'];
    	$hosts = array();
    	foreach ($server_ids as $k=>$v) {

    		if (!isset($conf['hosts'][$v])) {
    			trigger_error('别名为['.$v.']的队列服务器在配置文件里没有定义!', E_USER_ERROR);
    		}

    		$hosts[] = $conf['hosts'][$v];
    	}

    	$this->hosts = $hosts;
    }

    public function __destruct() {

    	$this->disconnect();
    }

    /**
     * 连接队列服务器
     *
     * @param boolean persistent true/false, default:false
     */
    public function connect($persistent=false, $database=15) {

    	$host = $this->hosts[0];

    	if (!isset($host['database'])) $host['database'] = $database;
    	$host['connection_persistent'] = $persistent;

    	/// 如果已连接，直接返回吖
    	///if ($this->connected) return true;

    	try {
    		$this->redis = new Predis_Client($host);
    		$this->connected = true;
    	}
    	catch(Exception $ex) {
    		$this->connected = false;

    		throw_exception('redis 队列服务连接失败');
    	}
    }

    /**
     * 关闭连接
     */
    public function disconnect() {

    	if (is_object($this->redis)) {
    		$this->connected = false;
    		$this->redis->disconnect();
    	}
    }

    /**
     * 将文本信息放入一个队列
     *  如果入队列成功，返回布尔值：true
     *  如果如队列失败，返回布尔值：false
     *
     * @param string $data
     */
    public function put($data) {

    	$this->connect();

    	$ret = $this->redis->lpush($this->real_q_name, $data);

    	$this->disconnect();

    	if ($ret) return true;
    	else return false;
    }

    /**
     * 警告！高级方法，如果你对队列的机制不是完全的了解，请不要使用！
     *
     * 将文本信息追加的队列的尾部
     *  如果入队列成功，返回布尔值：true
     *  如果如队列失败，返回布尔值：false
     *
     * @param string $data
     */
    public function append($data) {

    	$this->connect();

    	$ret = $this->redis->rpush($this->real_q_name, $data);

    	$this->disconnect();

    	if ($ret) return true;
    	else return false;
    }

    /**
     * 从一个队列中取出文本信息
     *  返回该队列的内容
     *  如果没有未被取出的队列，则返回文本信息：HTTPSQS_GET_END
     */
    public function get() {

    	$this->connect();

    	$ret = $this->redis->rpop($this->real_q_name);

    	$this->disconnect();

    	if (is_null($ret)) return 'HTTPSQS_GET_END';

    	return $ret;
    }

	/**
	 * 查看队列状态
	 */
    public function status() {

    	$this->connect();

    	$ret = $this->redis->info();

    	$this->disconnect();

    	return $ret;
    }

    /**
     * persistent put
     * 将文本信息放入一个队列
     *  如果入队列成功，返回布尔值：true
     *  如果如队列失败，返回布尔值：false
     *
     * @param string $data
     */
    public function pput($data) {

    	$this->connect(true);

    	$ret = $this->redis->lpush($this->real_q_name, $data);

    	if ($ret) return true;
    	else return false;
    }

    /**
     * 警告！高级方法，如果你对队列的机制不是完全的了解，请不要使用！
     *
     * 将文本信息追加的队列的尾部
     *  如果入队列成功，返回布尔值：true
     *  如果如队列失败，返回布尔值：false
     *
     * @param string $data
     */
    public function pappend($data) {

    	$this->connect(true);

    	$ret = $this->redis->rpush($this->real_q_name, $data);

    	if ($ret) return true;
    	else return false;
    }

    /**
     * persistent get
     * 从一个队列中取出文本信息
     *  返回该队列的内容
     *  如果没有未被取出的队列，则返回文本信息：HTTPSQS_GET_END
     */
    public function pget() {

    	$this->connect(true);

    	$ret = $this->redis->rpop($this->real_q_name);

    	if (is_null($ret)) return 'HTTPSQS_GET_END';

    	return $ret;
    }

	/**
	 * persistent status
	 * 查看队列状态
	 */
    public function pstatus() {

    	$this->connect(true);

    	return $this->redis->info();
    }
}
?>